#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <mysql/mysql.h>
#include "fileUtil.h"
#include "config.h"
#include "configField.h"
#include<sys/time.h>
#include "linkQueue.h"
#include <unistd.h>
#include <signal.h>
#include <stdbool.h>
#include <asm/errno.h>


//static MYSQL *mysqlClient = NULL;
static char *baseSqlFormat = NULL;
static LinkQueue *queue;
static bool flag=true;

pthread_t produceThread;
pthread_t *consumeThreads;
static MYSQL *mysqlConnects = NULL;

int initMysqlConnect() {
    printf("init mysql connection\n");
    const char *clientInfo = mysql_get_client_info();
    printf("mysql client: %s\n", clientInfo);
    mysqlConnects = calloc(threads,sizeof(MYSQL));

    char reConnect=1;
    int i=0;
    for (i = 0; i < threads; ++i) {
        mysql_init(mysqlConnects+i);
        mysql_options(mysqlConnects+i,MYSQL_OPT_RECONNECT,&reConnect);
        if (mysql_real_connect(mysqlConnects+i, mysqlHost, mysqlUser, mysqlPassword, mysqlDB, mysqlPort, NULL, 0) == NULL) {
            printf("failed connect to mysql,%s\n", mysql_error(mysqlConnects+i));
            return 1;
        }

    }
    printf("init mysql connections success\n");
    return 0;

}

void closeMysqlConnect() {
    printf("closing mysql client\n");
    int i=0;
    for (i = 0; i < threads; ++i) {
        mysql_close(mysqlConnects+i);

    }
    free(mysqlConnects);
    mysql_library_end();
}



char *stringAppend(char *src, char *str, int *cap) {

    if (*cap <= (strlen(src) + strlen(str))) {
        *cap = *cap + strlen(str) * 2;
        char *resrc = realloc(src, *cap);
        if (resrc == NULL) {
            printf("realloc failed!\n");
            return src;
        }

        src = resrc;
    }
    strcpy(src + strlen(src), str);

    return src;

}


char *prepareSqlFormat() {
    baseSqlFormat = calloc(1024,sizeof(char));
    char *insertOperate = "insert into ";
    char *values = "values";
    int *cap = calloc(1,sizeof(int));
    *cap = 20;
    baseSqlFormat = stringAppend(baseSqlFormat, insertOperate, cap);
    baseSqlFormat = stringAppend(baseSqlFormat, mysqlTable, cap);
    baseSqlFormat = stringAppend(baseSqlFormat, "(", cap);
    baseSqlFormat = stringAppend(baseSqlFormat, csvColumn, cap);
    baseSqlFormat = stringAppend(baseSqlFormat, ") ", cap);
    baseSqlFormat = stringAppend(baseSqlFormat, values, cap);
    baseSqlFormat = stringAppend(baseSqlFormat, "(", cap);
    baseSqlFormat = stringAppend(baseSqlFormat, "%s", cap);
    baseSqlFormat = stringAppend(baseSqlFormat, ");", cap);
    free(cap);
    return baseSqlFormat;
#ifdef DEBUG
    printf("exit prepareSqlFormat:%s\n", baseSqlFormat);
#endif


}



char *quoteValues(char *line) {
#ifdef DEBUG
    printf("enter main.quoteValues,%s\n",line);
#endif
    char *buffer = calloc(500, sizeof(char));
    int len = strlen(line);

    int index = 0;
    buffer[index++] = '\'';
    int i = 0;
    for ( i = 0; i < len; ++i) {
        char c = *(line + i);
        if (c == ',') {
            buffer[index++] = '\'';
            buffer[index++] = c;
            buffer[index++] = '\'';
        } else {
            buffer[index++] = c;
        }
    }
    buffer[index] = '\'';
#ifdef DEBUG
    printf("exit main.quoteValues,%s\n",buffer);
#endif
    return buffer;
}

void executeSql(char *sql,MYSQL *mysqlConnect) {
#ifdef DEBUG
    printf("insert sql:%s\n", sql);
#endif
    mysql_ping(mysqlConnect);
    int result = mysql_query(mysqlConnect, sql);
    if (result) {
        printf("insert failed,%s\n", mysql_error(mysqlConnect));
    }
}



void handleLine(char *line,MYSQL *mysqlConnect) {
#ifdef DEBUG
    printf("main.handleLine:%s\n", line);
#endif

    char *valueBuffer = quoteValues(line);
    char *sb=calloc(1024,sizeof(char));
    sprintf(sb, baseSqlFormat, valueBuffer);
    executeSql(sb,mysqlConnect);
    free(valueBuffer);
    free(sb);
}

void lineToQueue(char *line) {
    char *newline=calloc(strlen(line)+1,sizeof(char));
    strcpy(newline,line);
    enQueue(queue,newline);
}

long gettimeMS() {
    struct timeval curr;
    gettimeofday(&curr, NULL);
    return curr.tv_sec*1000 + curr.tv_usec/1000;
}


void *getLineFromQueue(void *arg){
    int j = 0;
    while(flag){
        char *data = deQueue(queue);
        if (data != NULL) {
            int len = strlen(data);
            if (len > 0 && data[len - 1] == '\n') {
                data[len - 1] = '\0';
            }
//            printf("get [%s] from queue\n", data);
            handleLine(data,(MYSQL*)arg);
            free(data);
            j++;
            if(j%100==0){
                pthread_t self = pthread_self();
                printf("thread[%ld] process %d lines \n",self,j);
            }
        }
    }
}

void *readLinesToQueue(void *args){
    readFile(csvFile, lineToQueue);
}


void shutdownHook(int sign){
    flag = 0;
    if(pthread_kill(produceThread,0)==0){
        pthread_cancel(produceThread);
    }
    if(consumeThreads != NULL){
        int i =0;
        for ( i = 0; i < threads; ++i) {
            pthread_cancel(consumeThreads[i]);
        }
        free(consumeThreads);
    }

    freeConfig();
    closeMysqlConnect();
    if(baseSqlFormat){
        free(baseSqlFormat);
    }

    freeQueue(queue);
    printf("application shutdown at %ld\n",gettimeMS());
    exit(0);
}


int main(int argc, char *argv[]) {
    signal(SIGTERM,shutdownHook);
    signal(SIGINT,shutdownHook);
    long t1= gettimeMS();
    printf("application starting %ld\n",t1);

    parseConfig(argc,argv);
    if(initMysqlConnect()!=0){
        return 1;
    }
    prepareSqlFormat();

    queue = newQueue(500);
    consumeThreads = calloc(threads,sizeof(pthread_t));
    pthread_create(&produceThread, NULL, readLinesToQueue, NULL);
    int i = 0;
    for ( i = 0; i < threads; ++i) {
        pthread_create(consumeThreads+i, NULL, getLineFromQueue, mysqlConnects+i);
    }


    pthread_join(produceThread,NULL);
    for ( i = 0; i < threads; ++i) {
        pthread_join(consumeThreads[i],NULL);
    }


}


